package com.sanmubird.mywebframework.eventbus.core;

import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

public abstract class AbstractMqConsumer<T> implements MqConsumer<T> {
    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractMqConsumer.class);

    protected static final long TIME_OUT = 30;

    @SuppressWarnings("InfiniteLoopStatement")
    @Override
    public void run() {
        MqExchangeEnum mqExchangeEnum = getMqRouterEnum();
        LOGGER.info("MqConsumer Thread【{}】started ...", mqExchangeEnum.getRoutingKey());
        List<MqObserver> observers = getMqObserverList();
        if (CollectionUtils.isEmpty(observers)) {
            throw new RuntimeException("No observer found for " + mqExchangeEnum.getRoutingKey());
        }

        while (true) {
            String value = fetchMessage();
            if (value == null) {
                continue;
            }
            LOGGER.info("RECEIVED MqMessage【 {} 】, detail: {}", mqExchangeEnum.getRoutingKey(), value);

            MqMessage<T> message = convertToMqMessage(value);

            observers.forEach(
                    o -> CompletableFuture.runAsync(() -> o.observe(message))
                            .exceptionally(
                                    (e) -> {
                                        LOGGER.error("{} FAILED Handle MqMessage【 {} 】, detail: {}", o.getClass().getSimpleName(), mqExchangeEnum.getRoutingKey(), value);
                                        e.printStackTrace();
                                        LOGGER.error(e.getMessage());
                                        String stackTraceStr = Arrays.stream(e.getStackTrace()).map(StackTraceElement::toString).collect(Collectors.joining("\n"));
                                        LOGGER.error(stackTraceStr);
                                        throw new RuntimeException(e);
                                    }
                            )
            );
        }
    }

}
